package com.bds.snatch.system.queue;

import com.bds.snatch.model.TemplateUrl;
import com.bds.snatch.system.task.ParseHtmlTask;
import com.bds.snatch.system.util.Const;
import com.bds.snatch.system.util.RedisUtil;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.List;

@Service
public class FetchHtmlConsumer extends Thread{
    private static Logger logger= LoggerFactory.getLogger(FetchHtmlConsumer.class);
    /**
     * 监控redis
     * 阻塞队列，10s超时
     */
    @Override
    public void run() {
        while(true) {
            byte[] bytes= null;
            try {
                bytes = Const.TEMPLATE_URLS_QUEUES.getBytes("utf-8");
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            }
            long urls_count= RedisUtil.getInstance().llength(bytes);
            //队列任务数量大于10时，才使用多线程处理Html
            logger.info("---------------从Redis队列里获取的TemplateUrl:---------------" + urls_count );
            //阻塞队列，假死-->>改用休眠
            if(urls_count==0){
                try {
                    sleep(1500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                continue;
            }else if(urls_count>Const.MAX_QUEUE_SIZE){
                List<byte[]> list = RedisUtil.getInstance().lpopList(bytes);
                List<TemplateUrl>templs=new ArrayList<>();
                for(byte[] templ:list) {
                    String temp=new String(templ);
                    JSONObject jsonObject = JSONObject.parseObject(temp);
                    TemplateUrl templateUrl = (TemplateUrl) JSONObject.toJavaObject(jsonObject, TemplateUrl.class);
                    templs.add(templateUrl);
                }
                    logger.info("---------------多线程处理Html....---------------");
                    //根据任务数量确定开几个线程（单次任务量多时需修改常量PER_THREAD_HTMLS）
                    int thread_count = templs.size() / Const.PER_THREAD_HTMLS;
                    if (thread_count == 0) {
                        new ParseHtmlTask(templs).start();
                    } else {
                        for (int i = 0; i < thread_count; i++) {
                            int start = i * Const.PER_THREAD_HTMLS;
                            int end = 0;
                            if (i == thread_count - 1) {
                                end = templs.size();
                            } else {
                                end = start + Const.PER_THREAD_HTMLS;
                            }
                            List<TemplateUrl> templateUrls = templs.subList(start, end);
                            new ParseHtmlTask(templateUrls).start();
                        }
                    }
            }else {
                //队列任务数量大于10时,阻塞队列，10s超时
                List<byte[]> list = RedisUtil.getInstance().brpop(bytes);
                if (list != null && list.size() > 1) {
                    String templArray = new String(list.get(1));
                    logger.info("---------------从Redis队列里获取的TemplateUrl:---------------" + templArray);
                    if(StringUtils.isNoneEmpty(templArray)||"0".equals(templArray)) {
                        JSONObject jsonObject = JSONObject.parseObject(templArray);
                        TemplateUrl templateUrl = (TemplateUrl) JSONObject.toJavaObject(jsonObject, TemplateUrl.class);
                        new ParseHtmlTask(templateUrl).start();
                    }
                } else {
                    System.out.println("Redis队列为空,没有“解析html”任务。。。");
                }
            }
        }
    }

}

